/*
 * Copyright (c) 2019-2029, Dreamlu 卢春梦 (596392912@qq.com & dreamlu.net).
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package netty;

import net.dreamlu.iot.mqtt.codec.ByteBufferUtil;
import net.dreamlu.iot.mqtt.core.server.broker.MqttBrokerMessageListener;
import net.dreamlu.iot.mqtt.core.server.dispatcher.IMqttMessageDispatcher;
import net.dreamlu.iot.mqtt.core.server.event.IMqttMessageListener;
import net.dreamlu.iot.mqtt.core.server.model.Message;
import net.dreamlu.iot.mqtt.core.server.support.DefaultMqttMessageDispatcher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;

/**
 * mqtt服务器
 *
 * @author zhangtao
 * @date 2021/12/16
 */
public class MqttBroker {
    private static final Logger logger = LoggerFactory.getLogger(MqttBroker.class);

    public static void main(String[] args) {
        // 1. 消息转发处理器，可用来实现集群
        IMqttMessageDispatcher messageDispatcher = new DefaultMqttMessageDispatcher();
        // 2. mqtt broker 消息转发处理
        IMqttMessageListener messageListener = new MqttBrokerMessageListener(messageDispatcher);

        IMqttMessageListener customListener = new IMqttMessageListener() {
            @Override
            public void onMessage(ChannelContext context, String clientId, Message message) {
                messageListener.onMessage(context, clientId, message);
                logger.error("topic:{} message:{}", message.getTopic(), ByteBufferUtil.toString(message.getPayload()));

//                if ("/api/adapter/dash".equals(message.getTopic())) {
//                    logger.info("topic:{} message:{}", message.getTopic(), ByteBufferUtil.toString(message.getPayload()));
//                }
            }
        };
        // 3. 启动服务
        net.dreamlu.iot.mqtt.core.server.MqttServer.create()
                .ip("0.0.0.0")
                .port(1883)
                .readBufferSize(1024 * 1024 * 4)
                .maxBytesInMessage(1024 * 1024 * 4)
                .messageDispatcher(messageDispatcher)
                .messageListener(customListener)
                .httpEnable(true)
//                .debug()
                .start();
    }
}
